from .watermark import WatermarkGenerator  # 导入水印生成器模块
from .windows import EventTimeTumblingWindow  # 导入事件时间滚动窗口模块
from collections import defaultdict  # 导入默认字典模块

class Node:
    def __init__(self, name, func):
        """初始化节点"""
        self.name = name  # 节点名称
        self.func = func  # 节点对应的处理函数
        self.children = []  # 子节点列表

    def add_child(self, child_node):
        """添加子节点到当前节点的children列表"""
        self.children.append(child_node)  # 将子节点添加到列表中
        return self  # 返回当前节点，支持链式调用

class DAG:
    def __init__(self, window_size=300, allowed_lateness=5):
        """初始化有向无环图（DAG）"""
        self.nodes = {}  # 存储所有节点的字典
        self.root = None  # 根节点
        self.watermark_gen = WatermarkGenerator(max_lag=allowed_lateness)  # 初始化水印生成器
        self.window = EventTimeTumblingWindow(window_size, allowed_lateness)  # 初始化事件时间滚动窗口
        self.word_counts = defaultdict(int)  # 用于统计单词计数的默认字典
        self.processed_ids = set()  # 存储已处理的事件ID集合

    def add_node(self, name, func):
        """添加节点到DAG中"""
        node = Node(name, func)  # 创建新节点
        self.nodes[name] = node  # 将节点添加到节点字典中
        if not self.root:  # 如果根节点为空，则设置为当前节点
            self.root = node
        return node  # 返回新节点

    def add_edge(self, parent, child):
        """在父节点和子节点之间添加边"""
        self.nodes[parent].add_child(self.nodes[child])  # 将子节点添加到父节点的子节点列表中

    def execute(self, data):
        """执行DAG，处理输入数据"""
        self._execute_node(self.root, data)  # 从根节点开始递归执行

    # Reduce算子
    def _execute_node(self, node, record):
        """递归执行节点及其子节点"""
        if node.name == 'source':  # 如果节点是数据源节点
            word = record['word'].lower()  # 小写化单词，确保不区分大小写
            self.watermark_gen.update(record['timestamp'])  # 更新水印
            record['_watermark'] = self.watermark_gen.value  # 将水印值添加到记录中
            record['_event_id'] = hash(f"{record['timestamp']}-{record['word']}")  # 生成唯一事件ID

        if node.name == 'window_operator':  # 如果节点是窗口操作节点
            self.window.add_record(record[1])  # 将记录添加到窗口中
            triggered = self.window.check_trigger(record[1]['_watermark'])  # 检查是否触发窗口计算
            for result in triggered:  # 遍历触发的结果
                self._process_window_data(result)  # 处理窗口数据
        if node.name == 'map_operator':  # 如果节点是映射操作节点
            record = node.func(record)  # 对记录执行映射函数

        elif node.name == 'keyby_operator':  # 如果节点是分组操作节点
            # record['_key'] = node.func(record)  # 对记录执行分组函数并生成分组键
            record = (node.func(record), record)

        elif node.name == 'reduce_operator':  # 如果节点是聚合操作节点
            # self.word_counts[record['_key']] += record.get('count', 1)  # 更新单词计数
            # record['total'] = self.word_counts[record['_key']]  # 将总计数添加到记录中
            record=record[1]
            result =node.func(record)

            # self.word_counts[result['word']] += result['count']  # 按 key 汇总计数

            record = {'key' :result['word'], 'count':result['count'],'_event_id':record['_event_id'],'window_end':record['_watermark'] }


        elif  'sink_operator' in node.name:  # 如果节点是输出节点
            if record['_event_id'] not in self.processed_ids:  # 如果事件ID未被处理过
                node.func(record)  # 执行输出函数
                self.processed_ids.add(record['_event_id'])  # 将事件ID标记为已处理


        for child in node.children:  # 遍历当前节点的子节点
            self._execute_node(child, record)  # 递归执行子节点

    def _process_window_data(self, data):
        """处理窗口数据并输出结果"""
        window_time_str = data['window_end'].strftime('%Y-%m-%d %H:%M:%S')  # 格式化窗口结束时间
        for result in data['results']:  # 遍历窗口结果
            print(f"[Window {window_time_str}] {result['word']}: {result['total']}")  # 输出窗口统计结果